package com.ypf.rpc.manager.client;


import com.ypf.rpc.message.RpcQueryRegisterRequestMessage;
import com.ypf.rpc.message.RpcRequestMessage;
import com.ypf.rpc.nettyHandler.HeartBeatClientHandler;
import com.ypf.rpc.nettyHandler.RpcQueryRegisterResponseMessageHandler;
import com.ypf.rpc.nettyHandler.RpcResponseMessageHandler;
import com.ypf.rpc.procotol.MessageCodecSharable;
import com.ypf.rpc.procotol.ProcotolFrameDecoder;
import com.ypf.rpc.procotol.SequenceIdGenerator;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 管理 Rpc客户端的channel
 */
@Slf4j
public class RpcClientManager {

    // 单例channel
    private static final Bootstrap bootstrap;

    static NioEventLoopGroup group;


    // Rpc 服务调用 PROMISE 集合
    public static Map<Integer, Promise<Object>> PROMISES;

    // 向注册中心查询 服务ip+端口号 的 Promise 集合
    public static Map<Integer, Promise<Object>> queryRegisterPROMISES;

    // Rpc服务调用 Channel集合 可能请求多个ip地址或端口号
    public static Map<String,Channel> channels;

    private static Channel queryRegisterChannel;

    private static final Object LOCK = new Object();

    static {
        bootstrap = new Bootstrap();
        group = new NioEventLoopGroup();
        initQueryRegisterChannel();
        initRpcChannel();
        channels = new ConcurrentHashMap<>();
        PROMISES = new ConcurrentHashMap<>();
        queryRegisterPROMISES = new ConcurrentHashMap<>();
    }

    public RpcClientManager(){
        group = new NioEventLoopGroup();
    }


    /**
     * 获取channel 没有channel就建立连接
     * @param inetSocketAddress
     * @return
     */
    public static Channel get(InetSocketAddress inetSocketAddress){
        String key = inetSocketAddress.toString();
        // 判断是否存在
        if (channels.containsKey(key)){
            Channel channel = channels.get(key);
            if (channel!=null&&channel.isActive()){  // 判断channel是不是不为空是否活着
                return channel;
            }
            channels.remove(key);
        }

        // 创建连接
        Channel channel = null;
        try {
            synchronized (LOCK){
                channel = bootstrap.connect(inetSocketAddress).sync().channel();
                channel.closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
                    @Override
                    public void operationComplete(Future<? super Void> future) throws Exception {
                        log.debug("断开连接");
                    }
                });
            }
        } catch (InterruptedException e) {
            channel.close();
            log.debug("连接客户端出错"+e);
            return null;
        }

        channels.put(key,channel);
        return channel;
    }

    private static void initQueryRegisterChannel() {
        // 消息处理handler
        MessageCodecSharable MESSAGE_CODE = new MessageCodecSharable();

        // 自定义查询服务返回消息处理器
        RpcQueryRegisterResponseMessageHandler RPC_QUERY_REGISTER_RESPONSE_HANDLER = new RpcQueryRegisterResponseMessageHandler();

        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            ChannelFuture channelFuture = new Bootstrap().group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 定长解码器
                            socketChannel.pipeline().addLast(new ProcotolFrameDecoder());
                            socketChannel.pipeline().addLast(MESSAGE_CODE);
                            socketChannel.pipeline().addLast(RPC_QUERY_REGISTER_RESPONSE_HANDLER);
                        }
                    }).connect(new InetSocketAddress("127.0.0.1",10001)).sync();

            queryRegisterChannel = channelFuture.channel();

//            RpcQueryRegisterRequestMessage rpcRegisterRequestMessage = new RpcQueryRegisterRequestMessage();
//            rpcRegisterRequestMessage.setServerName("com.ypf.rpc.service.HelloService2Impl");
//            queryRegisterChannel.writeAndFlush(rpcRegisterRequestMessage);

            // 异步等待关闭 此线程将不会在这阻塞
            queryRegisterChannel.closeFuture().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    System.out.println("客户端查询服务通道关闭。。。");
                    group.shutdownGracefully();
                    queryRegisterChannel = null;
                }
            });

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void initRpcChannel(){

        // 日志handler
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);

        // 消息处理handler
        MessageCodecSharable MESSAGE_CODE = new MessageCodecSharable();

        // 处理相应的handler
        RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();

        // 心跳发送处理器
        HeartBeatClientHandler HEARTBEAT_CLIENT = new HeartBeatClientHandler();

        bootstrap.channel(NioSocketChannel.class)
                .group(group)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new IdleStateHandler(0,20,0, TimeUnit.SECONDS));
                        // 定长解码器
                        socketChannel.pipeline().addLast(new ProcotolFrameDecoder());
                        socketChannel.pipeline().addLast(MESSAGE_CODE);
                        socketChannel.pipeline().addLast(LOGGING_HANDLER);
                        socketChannel.pipeline().addLast(HEARTBEAT_CLIENT);
                        socketChannel.pipeline().addLast(RPC_HANDLER);
                    }
                });
    }



    public void sendRpcRequest(RpcRequestMessage msg) throws InterruptedException {

        // 1 向服务中心查询 服务ip 和端口号
        RpcQueryRegisterRequestMessage queryMessage = new RpcQueryRegisterRequestMessage();
        int querySequenceId = SequenceIdGenerator.nextId();
        queryMessage.setSequenceId(querySequenceId);
        queryMessage.setServerName(msg.getInterfaceName());


        // 2 准备一个空 Promise 对象 来接收查询结果     指定 promise 对象异步接收结果线程
        DefaultPromise<Object> queryPromise = new DefaultPromise<Object>(group.next());

        // 3 存入 queryRegisterPROMISES
        queryRegisterPROMISES.put(querySequenceId,queryPromise);

        // 判断 查询通道是否为空 或者已经失效
        if (queryRegisterChannel==null||!queryRegisterChannel.isActive()){
            initQueryRegisterChannel();  // 重新初始化 queryRegisterChannel
        }

        // 4 发送消息
        queryRegisterChannel.writeAndFlush(queryMessage);

        // 5 等待查询结果
        queryPromise.await();


        InetSocketAddress service = null;

        if (queryPromise.isSuccess()){
            service = (InetSocketAddress) queryPromise.getNow();
        }else {
            // 调用失败
            throw new RuntimeException(queryPromise.cause());
        }

        // 查询到服务地址后 开始正式调用服务
        Channel channel = get(service);
        if (!channel.isActive()||!channel.isRegistered()){
            group.shutdownGracefully();
            return;
        }

        channel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()){
                log.debug("客户端发送消息成功");
            }
        });
    }

}
